/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data625.handler.datadispatch;

import cn.ac.iie.ulss.data625.commons.RuntimeEnv;
import static cn.ac.iie.ulss.data625.server.Data625StartUp.fileNames;
import static cn.ac.iie.ulss.data625.server.Data625StartUp.fu;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Random;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.log4j.Logger;

/**
 *
 * @author Li Mingyang
 */
public class DataWriter {
    
    private Schema grSingleDocSchema = null;
    private DatumReader<GenericRecord> singleDocsReader = null;
    private SimpleDateFormat secondFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
    static Logger logger = Logger.getLogger(DataWriter.class.getName());
    
    public DataWriter() throws IOException {
        Protocol protocol = Protocol.parse(new File("dx.json"));
        grSingleDocSchema = protocol.getType("dx");
        singleDocsReader = new GenericDatumReader(grSingleDocSchema);
    }
    
    public void writeData(GenericArray docsSet) throws IOException {
        Iterator itor = docsSet.iterator();
        GenericRecord record = null;
        Random random = new Random();
        String split = "\",\"";
        Date d = new Date();
        while (itor.hasNext()) {
            StringBuilder sb = new StringBuilder();
            ByteArrayInputStream databis = new ByteArrayInputStream(((ByteBuffer) itor.next()).array());
            BinaryDecoder dataDecoder = new DecoderFactory().binaryDecoder(databis, null);
            record = (GenericRecord) singleDocsReader.read(null, dataDecoder);
            String time = record.get("C_FSSJ").toString();
            d.setTime(Long.parseLong(time) * 1000L);
            String fssj = this.secondFormat.format(d);
            String dxzx_id = new StringBuilder().append(record.get("C_JRDID").toString()).append("0").toString();
            String fwq_ip = ipveStr2long(record.get("C_YYSIP").toString());
            String dx_xh = record.get("C_DXXH").toString();
            String dxx_block_id = "20060101";
            String ydz = record.get("C_YDZ").toString();
            String mddz = record.get("C_MDDZ").toString();
            String dxnr = record.get("C_DXNR").toString().replaceAll("\n", "").replaceAll("\r", "").replaceAll("\"", "\"\"");
            String dxlx = getDxlx(Integer.parseInt(record.get("C_YDZHDLX").toString()),
                    Integer.parseInt(record.get("C_YDZYYSID").toString()),
                    Integer.parseInt(record.get("C_MDDZHDLX").toString()),
                    Integer.parseInt(record.get("C_MDDZYYSID").toString()));
            String bmlx = record.get("C_BMLX").toString();
            String dxfl = getDxfl(Integer.parseInt(record.get("c_sfwsb").toString()),
                    (GenericData.Array) record.get("tlv_fgz_pzids"),
                    (GenericData.Array) record.get("tlv_gz_pzids"),
                    (GenericData.Array) record.get("tlv_dsx_pzids"));
            String ydz_sd = record.get("C_YDZSSSID").toString();
            String mddz_sd = record.get("C_MDDZSSSID").toString();
            String fxr_szd = record.get("C_YDZDQSID").toString();
            String sxr_szd = record.get("C_MDDZDQSID").toString();
            String fx_dxzx = new StringBuilder().append(record.get("C_JRDID").toString()).append("0").toString();
            String sx_dxzx = "";
            sb.append("\"").append(fssj).append(split).
                    append(dxzx_id).append(split).append(fwq_ip).append(split).
                    append(dx_xh).append(split).append(dxx_block_id).append(split).
                    append(ydz).append(split).append(mddz).append(split).append(dxnr).
                    append(split).append(dxlx).append(split).append(bmlx).append(split).
                    append(dxfl).append(split).append(ydz_sd).append(split).append(mddz_sd).
                    append(split).append(fxr_szd).append(split).append(sxr_szd).append(split).append(fx_dxzx).
                    append(split).append(sx_dxzx).append("\"$EOF$");
            while (fileNames.size() != RuntimeEnv.getParamAsInteger(RuntimeEnv.FILE_NUM)) {
                try {
                    logger.info("");
                    Thread.sleep(30);
                    //如果不相同，那就说明FileControlThread正在调整目前的数据文件名，等20ms再看是不是一样了
                } catch (InterruptedException ex) {
                    logger.error(ex.getMessage(), ex);
                }
            }
            fu.WriteFile(sb.toString(), RuntimeEnv.getParamAsString(RuntimeEnv.FILE_PATH),
                    new StringBuilder().append(fileNames.get(random.nextInt(RuntimeEnv.getParamAsInteger(RuntimeEnv.FILE_NUM)))).toString(), "utf-8");
        }
    }
    
    private String getDxlx(int dx_ydz_hd_type, int dx_ydz_yys, int dx_mddz_hd_type, int dx_mddz_yys) {
        int dxlx = 0;
        if (dx_ydz_hd_type == 1 && dx_ydz_yys == 1 && dx_mddz_hd_type == 1 && dx_mddz_yys == 1) {
            dxlx = 101;
        } else if (dx_ydz_hd_type == 1 && dx_ydz_yys == 1 && dx_mddz_hd_type == 1 && dx_mddz_yys == 2) {
            dxlx = 102;
        } else if (dx_ydz_hd_type == 1 && dx_ydz_yys == 1 && dx_mddz_hd_type == 2) {
            dxlx = 103;
        } else if (dx_ydz_hd_type == 1 && dx_ydz_yys == 1 && dx_mddz_hd_type == 3) {
            dxlx = 104;
        } else if (dx_ydz_hd_type == 1 && dx_ydz_yys == 2 && dx_mddz_hd_type == 1 && dx_mddz_yys == 1) {
            dxlx = 201;
        } else if (dx_ydz_hd_type == 1 && dx_ydz_yys == 2 && dx_mddz_hd_type == 1 && dx_mddz_yys == 2) {
            dxlx = 202;
        } else if (dx_ydz_hd_type == 1 && dx_ydz_yys == 2 && dx_mddz_hd_type == 2) {
            dxlx = 203;
        } else if (dx_ydz_hd_type == 1 && dx_ydz_yys == 1 && dx_mddz_hd_type == 3) {
            dxlx = 204;
        } else if (dx_ydz_hd_type == 2 && dx_mddz_hd_type == 1 && dx_mddz_yys == 1) {
            dxlx = 301;
        } else if (dx_ydz_hd_type == 2 && dx_mddz_hd_type == 1 && dx_mddz_yys == 2) {
            dxlx = 302;
        } else if (dx_ydz_hd_type == 2 && dx_mddz_hd_type == 2) {
            dxlx = 303;
        } else if (dx_ydz_hd_type == 2 && dx_mddz_hd_type == 3) {
            dxlx = 304;
        } else if (dx_ydz_hd_type == 3 && dx_mddz_hd_type == 1 && dx_mddz_yys == 1) {
            dxlx = 401;
        } else if (dx_ydz_hd_type == 3 && dx_mddz_hd_type == 1 && dx_mddz_yys == 2) {
            dxlx = 402;
        } else if (dx_ydz_hd_type == 3 && dx_mddz_hd_type == 2) {
            dxlx = 403;
        } else if (dx_ydz_hd_type == 3 && dx_mddz_hd_type == 3) {
            dxlx = 404;
        } else {
            dxlx = 9999;
        }
        return String.valueOf(dxlx);
    }
    
    private String getDxfl(int sfwsb, GenericArray tlv_fgz_pzids, GenericArray tlv_gz_pzids, GenericArray tlv_dsx_pzids) {
        int c_pzlx = -1;
        int dxfl = 5;
        if (sfwsb == 1) {
            dxfl = 8;
        } else if ((tlv_fgz_pzids.isEmpty()) && (tlv_gz_pzids.isEmpty()) && (tlv_dsx_pzids.isEmpty())) {
            dxfl = 9;
        } else {
            HashSet<Integer> tSet = new HashSet<Integer>();
            for (Iterator it = tlv_fgz_pzids.iterator(); it.hasNext();) {
                Object o = it.next();
                GenericRecord tlv = (GenericRecord) o;
                c_pzlx = ((Integer) tlv.get("C_PZLX"));
                tSet.add(c_pzlx);
            }
            for (Iterator it = tlv_gz_pzids.iterator(); it.hasNext();) {
                Object o = it.next();
                GenericRecord tlv = (GenericRecord) o;
                c_pzlx = ((Integer) tlv.get("C_PZLX"));
                tSet.add(c_pzlx);
            }
            for (Iterator it = tlv_dsx_pzids.iterator(); it.hasNext();) {
                Object o = it.next();
                GenericRecord tlv = (GenericRecord) o;
                c_pzlx = ((Integer) tlv.get("C_PZLX"));
                tSet.add(c_pzlx);
            }
            if (tSet.contains(702)) {
                dxfl = 1;
            } else if (tSet.contains(701)) {
                dxfl = 2;
            } else if (tSet.contains(201)) {
                dxfl = 3;
            } else if (tSet.contains(1002)) {
                dxfl = 4;
            } else if (tSet.contains(1101)) {
                dxfl = 10;
            } else if (tSet.contains(1102)) {
                dxfl = 11;
            } else {
                dxfl = 5;
            }
        }
        return String.valueOf(dxfl);
    }
    
    private String ipveStr2long(String in) {
        if (in == null || in.equals("")) {
            return "-1";
        }
        String str = in.trim();
        long[] ip = {0, 0, 0, 0};
        int i = 0;
        int j = 0;
        int m = 1;
        char c = str.charAt(0);
        if (c > '9' || c < '0') {
            return "-1";
        }
        ip[0] = c - '0';
        for (i = 1; i < str.length(); i++) {
            c = str.charAt(i);
            if (c == '.') {
                if (m < 1 || ip[j] > 255) {
                    return "-1";
                }
                j++;
                m = 0;
            } else if (c > '9' || c < '0') {
                return "-1";
            } else {
                ip[j] = c - '0' + (ip[j] << 3) + (ip[j] << 1);
                m++;
            }
        }
        if (j < 3 || (j == 3 && m == 0)) {
            return "-1";
        }
        long ipl = (ip[0] << 24) | (ip[1] << 16) | (ip[2] << 8) | ip[3];
        return String.valueOf(ipl);
    }
}
